#include "table.h"
#include "serializer.h"
#include "offset_table.h"
#include "query_helper.h"
#include <errno.h>
#include <unistd.h>
#include <stdio.h>
#include "replication_ignore.h"
#include "monitor.h"
#undef MAX
#define MAX(a,b)        ((a) > (b) ? (a) : (b))
#undef MIN
#define MIN(a,b)        ((a) < (b) ? (a) : (b))
typedef long long int L;
typedef unsigned long long int UL;


map<string, Table*> Table::tables;
LOCK Table::tablesLock;
Table* Table::get(const string& table_name) {
//	LOG << "table_name = " << table_name << ENDL;
	SRLock _srl(tablesLock);
	Table* t = NULL;
	if(tables.find(table_name) == tables.end()) {
		t = new Table(table_name);
		tables[table_name] = t;
	} else {
		t = tables[table_name];
	}
	return t;
}
void Table::rename(const string& newname) {
	SRLock _srl(tablesLock);
	OffsetTable::rename(name, newname);
	tables[newname] = this;
	tables.erase(this->name);
	this->name = newname;
}

Table::Table(string table_name)  {
//	DB_INIT;
//	rowCount = 0;
	linebuf_size = 0;
	num_cols = 0;
	offsetTable = OffsetTable::Get(table_name, true);
	sprintf(logfile, "/tmp/sqlfslog/%s.log", table_name.c_str());
	if(::truncate(logfile, 0)) {
		fprintf(stderr, "WARNING: Can not truncate the log file %s ,errno = %d\n", logfile, errno);
	}
	logp = fopen(logfile, "a");
	name = table_name;
//	DB_START;
	QUERY
	Query tmpq = con->query();
	q << "DESCRIBE " << name;
	//this->log(q.str());
	UseQueryResult res = q.use();
	res.fetch_row();
//	DB_STOP;
	columns_str = "";
	//extra_bytes = 0;
	while(true) {
//		DB_START;
		Row row = res.fetch_row();
//		DB_STOP;
		if(!row)
			break;
		//columns.push_back((string)row[0]);
		if(columns_str == "") {
			columns_str += "(";
			columns_str += (string)row[0];
		} else {
			columns_str += ", ";
			columns_str += (string)row[0];
		}
		string s = (string)row[1];
		if(s.substr(0,3) == "int") {
			types[num_cols++] = INT;
		} else {
			types[num_cols++] = STRING;
		}
	}
	columns_str += ")";
	printf("num_cols = %lu\n", num_cols);
	//we need to ignore the rowid column
	//LOG << "columns_str = " << columns_str << ENDL;
}
/*void Table::write(const char* buf, int size, off_t offset) {
	DB_INIT;
	TREE_INIT;
	if(*buf == '\n' && last_line == "") {
		if(size == 1)
			return;
		--size;
		++buf;
	}

	last_line.append(buf, size);
	size_t pos;
	DB_START;
	QUERY
	q << "INSERT INTO " << name << columns_str << " VALUES ";
	DB_STOP;
	vector<size_t> rowSizes;
	size_t diffSize = 0;
	bool first = true;
	while((int)(pos = last_line.find_first_of('\n')) != -1) {
		string s1;
		if(pos >= 1 && last_line[pos - 1] == '\r') {
			//LOG <<"ignoring the " << last_line[pos-1] << " character" << ENDL;
			s1 = last_line.substr(0, pos - 1).c_str();
		} else {
			s1 = last_line.substr(0, pos);
		}

		if(s1 == "")
			continue;
		last_line = last_line.substr(pos + 1);
		//LOG << "last_line = " << last_line << ENDL;
		size_t rowSize = 0;
		size_t col_count = 0;
		DB_START;
		Query tmpq = con->query();
		tmpq << "(";
		DB_STOP;
		int comma_pos = 0;
		string token;
		while(col_count < num_cols && comma_pos != -1) {
			comma_pos = s1.find_first_of(',');
			//LOG << "comma_pos = " << comma_pos << ENDL;
			if(comma_pos == -1) {
				token = s1;
			} else {
				token = s1.substr(0, comma_pos);
			}
			s1 = s1.substr(comma_pos + 1);
			//LOG << "token = " << token << ENDL;
			if(token == "") {
				//LOG << "token is empty!!"  << ENDL;
				if(types[col_count].substr(0,3) == "int") {
					token = "NULL" ;
					//++extra_bytes;
					//LOG << "token changed" << ENDL;
				}
			}
			if(col_count > 0)
				tmpq << ", ";
			if(types[col_count].substr(0,3) != "int") {
				tmpq << quote << token ;
				rowSize += token.size() + 1;
			} else if(token != "NULL") {
				long long num = atoll(token.c_str());
				char intbuf[25];
				sprintf(intbuf, "%lld", num);
				tmpq << intbuf;
				rowSize += strlen(intbuf) + 1;
			} else {
				tmpq << "NULL";
				rowSize += 1;
			}
			++col_count;
		}
		tmpq << ")";
		if(col_count == num_cols) {
			if(first) {
				first = false;
			} else {
				q << ",";
			}
			q << tmpq.str();
			//LOG << "q = " << q.str() << ENDL;
			rowSizes.push_back(rowSize);
			diffSize += rowSize;
		} else {

		}
		tmpq.reset();
	}

	if(!first) {
		//LOG << "q before execution = " << q.str() << ENDL;
		//this->log(q.str());
		DB_START;
		q.exec();
		DB_STOP;
		//update of the db indexes will be handled by replication
		TREE_START;
		long long int rowid = offsetTable->getLastRow();
		for(size_t i = 0, len = rowSizes.size();  i < len; ++i) {
			//LOG << "adding row of size " << rowSizes[i] << " to offsetTable" << ENDL;
			offsetTable->appendRow(++rowid, rowSizes[i]);
			//LOG <<"current size of offsetTable = " << offsetTable->size << ENDL;
		}
		TREE_STOP;
		repl_ignore::add_insert(name, diffSize);
	}
	//Pool::Get()->release(con);
}*/

void append_to_q(Query& q, const char* str) {
	q << str;
}

void Table::handle_one_line(Query& q, char** bufptr, bool first) {
	char *rowbufptr = rowbuf;
	size_t col_count = 0;
	size_t rowSize = 0;
	char *ptr = *bufptr;
	char cur;
	if(!first) {
		*(++rowbufptr) = ',';
	}
	*(++rowbufptr) = '(';
	bool first_field = true;
	bool field_start = true;
	while(col_count < num_cols) {
		if(field_start) {
			if(types[col_count] == STRING) {
				*(++rowbufptr) = '\'';
			} else if(*ptr == ',' || *ptr == '\r' || *ptr == '\n') {
				*(++rowbufptr) = 'N';
				*(++rowbufptr) = 'U';
				*(++rowbufptr) = 'L';
				*(++rowbufptr) = 'L';
			}
		}
		cur = *ptr;
		field_start = false;
		if(cur == '\'') {
			*(++rowbufptr) = '\'';
		}

		if(cur ==',' || cur == '\r' || cur == '\n') {
			if(types[col_count] == STRING) {
				*(++rowbufptr) = '\'';
			}
			if(cur == ',')
				*(++rowbufptr) = ',';
			//after the work:
			++col_count;
			++rowSize;
			first_field = false;
			field_start = true;
			if(cur == '\n') {
				ptr = ptr + 1;
				break;
			}
			if(cur == '\r') {
				ptr = ptr + 2;
				break;
			}
		} else {
			*(++rowbufptr) = cur;
			++rowSize;
		}
		++ptr;
	}
	while(*(ptr - 1) != '\n') {
		++ptr;
	}
	for(unsigned int i = col_count; i < num_cols; ++i) {
		//add null ints or empty strings
		if(!first_field) {
			 *(++rowbufptr) = ',';
		}
		if(types[col_count]  == INT) {
			*(++rowbufptr) = 'N';
			*(++rowbufptr) = 'U';
			*(++rowbufptr) = 'L';
			*(++rowbufptr) = 'L';
		} else {
			*(++rowbufptr) = '\'';
			*(++rowbufptr) = '\'';
		}
	}
	//q << ")";
	*(++rowbufptr) = ')';
	*(++rowbufptr) = '\0';
	append_to_q(q, rowbuf + 1);
	//rowSizes.push_back(rowSize);
	//rowSizes[rowCount++] = rowSize;
	//assert(rowCount < 8192);
	long long int rowid = offsetTable->getLastRow();
	repl_ignore::add_insert(name, rowSize);
	offsetTable->appendRow(++rowid, rowSize);
	linebuf_size = ((*bufptr + linebuf_size) - ptr);
	*bufptr = ptr;
}
void Table::write(const char* buf, int size, off_t offset) {
	DB_INIT;
	TREE_INIT;
	if(*buf == '\n' && linebuf_size == 0) {
		if(size == 1)
			return;
		--size;
		++buf;
	}
	QUERY
	DB_START
	q << "INSERT INTO " << name << columns_str << " VALUES ";
	DB_STOP
	//vector<size_t> rowSizes;
	//rowCount = 0;
	bool first = true;
	char *bufptr = linebuf;
	while(true) {
		int cp_size = MIN((int)size, (int)(LINEBUF_SIZE - linebuf_size));
		//find the newline from reverse
		while(cp_size > 0 && *(buf + cp_size - 1) != '\n') --cp_size;
		if(cp_size > 0) {
			memcpy(linebuf + linebuf_size, buf, cp_size);
			linebuf_size += cp_size;
			size -= cp_size;
			buf += cp_size;
			while(linebuf_size != 0) { handle_one_line(q, &bufptr, first); first = false;}
			if(size == 0)
				break;
		} else {
			memcpy(linebuf + linebuf_size, buf, size);
			linebuf_size += size;
			break;
		}
	}

	if(!first) {
		this->log(q.str());
		q.exec();
	}
}

void Table::truncate(off_t size) {
	//last_line = "";
	linebuf_size = 0;
	QUERY
	DB_INIT;
	TREE_INIT;
	if(size == 0) {
		DB_START;
		q << "TRUNCATE " << name;
		DB_STOP;
		TREE_START;
		offsetTable->clear();
		TREE_STOP;
		repl_ignore::inc_truncate(name);
	} else if((UL)size >= offsetTable->getSize()) {
		return;
	} else {
		TREE_START;
		RowOffset ro = offsetTable->findRows(size, 0);
		TREE_STOP;
		DB_START;
		q << "SELECT * FROM " << name << " WHERE rowid >= " << ro.firstRow
				<< " AND rowid <= " << ro.lastRow << " ORDER BY rowid";
		cout << "q = " << q.str() << endl;
		UseQueryResult res = q.use();
		DB_STOP;
		unsigned long long tot = 0;
		unsigned long long firstRow = 0;
		while(true) {
			DB_START;
			Row row = res.fetch_row();
			DB_STOP;
			if(!row)
				break;
			tot += Serializer::RowSize(row, 1);
			if(tot >= ro.offset) {
				firstRow = (long long)row[0];
				break;
			}
		}

		if(firstRow == 0)
			firstRow = ro.lastRow;
		DB_START;
		while(Row row = res.fetch_row()) {};
		q.reset();
		q << "DELETE FROM " << name << " WHERE rowid > " << firstRow;
		DB_STOP;
		cout << " q = " << q.str() << endl;
		repl_ignore::add_delete(name, offsetTable->getSize() - tot);
		TREE_START;
		offsetTable->truncate(firstRow, tot);
		TREE_STOP;
		assert(offsetTable->getSize() == tot);
	}
	//this->log(q.str());
	DB_START;
	q.execute();
	DB_STOP;
	//this will be performed because of replication
}


size_t Table::read(char *buf, size_t size, off_t offset) {
	QUERY;
	RowOffset ro = offsetTable->findRows(offset, size);
	q << "SELECT * FROM " << name << " WHERE rowid >= " << ro.firstRow <<
			" AND rowid <=" << ro.lastRow << " ORDER BY rowid";
	Serializer::SerializeRow sr(buf, ro.offset, ro.size, 1);
	q.for_each(sr);
    return ro.size;
}

